package com.example.tableSql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}

/**
 * Connector Tables连接外部 系统创建表
 */
object Demo3 {

  case class Suppa(name: String, part: String, number: String, cost: String)

  def main(args: Array[String]): Unit = {

    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
    val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
    val filePath = "E:\\note\\flink-project\\temp\\supplier_data.csv"
    bsTableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("name", DataTypes.STRING())
        .field("part", DataTypes.STRING())
        .field("number", DataTypes.STRING())
      )
      .inAppendMode()
      .createTemporaryTable("MyTable")

    //    bsTableEnv.sqlQuery("select * from MyTable where number='5467'").execute().print()
    val resulttable: Table = bsTableEnv.sqlQuery("select * from MyTable where number='5467'")

    val value: String = bsTableEnv.explain(resulttable)
    println(value)



  }
}
